#include "vcl_stl.h"

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>
#include <netdb.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "h_base.h"
#include "h_xml.h"
using namespace h_base;

#include "simbat.h"
using namespace simbat;

#include "bkbase.h"
#include "bstcp.h"

struct HClusterNodeInfo
{
   int Id;
   HString Addr;
   int Port;

   HClusterNodeInfo()
      : Id(-1), Port(-1)
   {}
};

static char *SimbatBase = NULL;
static char ErrMsg[8192 + 1];

static const int COMM_TIME_OUT = 30000;
static int SockExec = -1;
static int MsgSock = -1;
static bool ExitFlag = false;

static HString SyslogDir;
static HSyslog Syslog;
static int LocalPort;
static HString ShrDir;
static HString LogDir;
static HString VarpoolFile;
static HString VarpoolLockFile;

static bool ClusterFlag = false;                 // 是否集群
static HNodeRole MyRole = nrMaster;              // 本服务器角色（MASTER/SLAVE/BACKUP）
static HString HotbackAddr;                      // 热备机IP地址
static int HotbackPort = -1;                     // 热备机端口
static bool HotbackEnableFlag = false;           // 是否启动热备
static bool HotbackPausedFlag = false;           // 是否暂停热备
static HLogLocation LogLocation = llLocal;       // log文件位置
static map<HString, HBackupInfo> BackupMap;      // 若干备份集群

static int MyId = -1;                            // 本服务器ID（从0开始）
static vector<HClusterNodeInfo> ClusterNodeVec;  // 集群配置信息

static void SignalHandler(int)
{
   ExitFlag = true;
}

static HString GetEnv(const HString &VarName)
{
   char *Env = getenv(VarName.c_str());
   HString Result;

   if ( Env == NULL )
      Result =  "";
   else
      Result =  Env;

   return Result;
}

static void LoadFromCfg(HXmlTree &Cfg)
{
   SyslogDir = EnvEvaluate(Cfg.GetValue("SYSLOG_DIR", "$SIMBAT_BASE/etc"));
   Syslog.Setup(SyslogDir, Cfg.GetValueUpper("SYSLOG_LEVEL", "INFO"));

   if ( !GetMyId(SimbatBase, MyId, MyRole, ErrMsg) )
   {
      Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
      return;
   }

   LocalPort = StrToInt(Cfg.GetValue("PORT").c_str());
   ShrDir = EnvEvaluate(Cfg.GetValue("SHR_DIR", "$SIMBAT_BASE/shr"));
   LogDir = EnvEvaluate(Cfg.GetValue("LOG_DIR", "$SIMBAT_BASE/log"));
   setenv("SIMBAT_SHR_DIR", ShrDir.c_str(), 1);
   setenv("SIMBAT_LOG_DIR", LogDir.c_str(), 1);

   {
      HString Str = Cfg.GetValueUpper("LOG_LOCATION", "LOCAL");
      if ( Str == "LOCAL" )
      {
         LogLocation = llLocal;
      }
      else
      {
         LogLocation = llNfs;
      }
   }

   {
      vector<HString> StrVec;
      Cfg.GetValues("CLUSTER", StrVec);
      ClusterFlag = StrVec.size() > 1;
      if ( ClusterFlag )
      {
         ClusterNodeVec.clear();
         for ( int i = 0; i < StrVec.size(); i++ )
         {
            HString &AddrStr = StrVec[i];
            vector<HString> AddrPort;
            SplitString(AddrStr.c_str(), ':', AddrPort, true);
            if ( AddrPort.size() != 2 && AddrPort.size() != 3 )
            {
               snprintf(ErrMsg, 8192, "Invalid cluster address(%s)", AddrStr.c_str());
               Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
               return;
            }
            HClusterNodeInfo Node;
            Node.Id = i;
            Node.Addr = AddrPort[0];
            Node.Port = StrToInt(AddrPort[1].c_str());
            if ( Node.Port <= 2048 )
            {
               snprintf(ErrMsg, 8192, "Port %s too small, must be greater than 2048", AddrStr.c_str());
               Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
               return;
            }
            ClusterNodeVec.push_back(Node);
         }
      }
      else
      {
         HClusterNodeInfo Node;
         Node.Id = 0;
         Node.Addr = "127.0.0.1";
         Node.Port = LocalPort;
         ClusterNodeVec.push_back(Node);
      }
   }

   HotbackEnableFlag = ( Cfg.GetValueUpper("HOTBACK_ENABLE", "FALSE") == "TRUE" );
   HotbackPausedFlag = ( Cfg.GetValueUpper("HOTBACK_PAUSED", "FALSE") == "TRUE" );
   HotbackAddr = "";
   HotbackPort = -1;

   HString AddrStr = Cfg.GetValue("HOTBACK_ADDR");

   if ( AddrStr == "" )
   {
      HotbackEnableFlag = false;
   }
   else
   {
      vector<HString> AddrPort;
      SplitString(AddrStr.c_str(), ':', AddrPort, true);
      if ( AddrPort.size() != 2 )
      {
         snprintf(ErrMsg, 2048, "Invalid hot-backup address(%s)", AddrStr.c_str());
         Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
      }
      HotbackAddr = AddrPort[0];
      HotbackPort = StrToInt(AddrPort[1].c_str());
   }

   LoadBackupInfoFromCfg(Syslog, Cfg, BackupMap);
}

static bool LoadCfg(HString &ErrStr)
{
   char FileName[500 + 1];
   HXmlTree Cfg;

   {
      HFileLockRead Lock("$SIMBAT_BASE/tmp/config.lck");
      snprintf(FileName, 500, "%s/etc/config.xml", SimbatBase);
      if ( !Cfg.LoadFromFile(FileName) )
      {
         ErrStr = Cfg.GetErrorString();
         return false;
      }
   }

   LoadFromCfg(Cfg);
   return true;
}

static int System(const char *Command)
{
   int r;
   pid_t pid;

   pid = fork();

   if ( pid == -1 )
   {
      return -1;
   }

   if ( pid == 0 )
   {
      BkSignal(SIGINT, SIG_DFL, 0);
      execl("/bin/sh", "sh", "-c", Command, NULL);
      exit(8);
   }

   waitpid(pid, &r, 0);
   return r;
}

static int System2(const char *Command)
{
   void (*OldAction)(int) = BkSignal(SIGCHLD, SIG_DFL, 0);

   int r = system(Command);
   int r2;

   for ( ; waitpid(-1, &r2, WNOHANG) > 0; )
      ;

   BkSignal(SIGCHLD, OldAction, 0);
   return r;
}

static void LoadVarpool()
{
   snprintf(ErrMsg, 8192, "LoadVarpool()");
   Syslog.Log(__FILE__, __LINE__, llInfo, "OTHER", ErrMsg);

   int r;
   HXmlTree Xml;
   vector<HXmlTree*> Nodes;

   if ( FileExist(VarpoolFile) )
   {
      HFileLockRead Lock(VarpoolLockFile.c_str());
      if( !Xml.LoadFromFile(VarpoolFile.c_str(), false) )
      {
         snprintf(ErrMsg, 8192, "Load var_pool fail, %s", Xml.GetErrorString().c_str());
         Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
         return;
      }
   }

   Xml.GetChildren(NULL, Nodes);

   for ( int i = 0; i < Nodes.size(); i++ )
   {
      HString Name = Nodes[i]->GetTag();
      HString Value = Nodes[i]->GetData();
      if ( setenv(Name.c_str(), Value.c_str(), 1) == -1 )
      {
         snprintf(ErrMsg, 8192, "setenv(%s,%s) fail, errno = %d", Name.c_str(), Value.c_str(), errno);
         Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
         break;
      }
   }
}

static void SetEnv(const char *Name, const char *Value)
{
   snprintf(ErrMsg, 8192, "setenv(%s,%s)", Name, Value);
   Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);

   if ( setenv(Name, Value, 1) == -1 )
   {
      snprintf(ErrMsg, 8192, "setenv(%s,%s) fail, errno = %d", Name, Value, errno);
      Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
   }
}

static void BackupLog(const char *BatchTime, const char *JobName, long &Offset, bool RestoreFlag)
{
   char FileName[500 + 1];

   snprintf(FileName, 500, "%s/log_%s/%s%s", LogDir.c_str(), BatchTime,
      JobName, RestoreFlag ? "_restore" : "");

   long Len = BkFileSize(FileName);

   if ( Len <= Offset )
   {
      return;
   }

   snprintf(FileName, 500, "log_%s/%s%s", BatchTime,
      JobName, RestoreFlag ? "_restore" : "");

   if ( ClusterFlag && LogLocation == llLocal )
   {
      HClusterNodeInfo &Node = ClusterNodeVec[( MyId + 1 ) % ClusterNodeVec.size()];
      SendBackupFile(Syslog, ShrDir, LogDir, SimbatBase, Node.Addr.c_str(), Node.Port, boJobLog, FileName, Offset, Len - Offset);
   }

   if ( MyRole == nrMaster && HotbackEnableFlag && !HotbackPausedFlag )
   {
      SendBackupFile(Syslog, ShrDir, LogDir, SimbatBase, HotbackAddr.c_str(), HotbackPort, boJobLog, FileName, Offset, Len - Offset);
   }

   for ( map<HString, HBackupInfo>::iterator Iter = BackupMap.begin();
         Iter != BackupMap.end(); Iter++ )
   {
      HBackupInfo &Backup = Iter->second;
      if ( Backup.EnableFlag && !Backup.PausedFlag && Backup.LogFlag )
      {
         vector<HString> AddrPort;
         SplitString(Backup.AddrVec[MyId % Backup.AddrVec.size()].c_str(), ':', AddrPort, true);
         HString Addr = AddrPort[0];
         int Port = StrToInt(AddrPort[1].c_str());
         SendBackupFile(Syslog, ShrDir, LogDir, SimbatBase, Addr.c_str(), Port, boJobLog, FileName, Offset, Len - Offset);
      }
   }

   Offset = Len;
}

void RunIt(const HExecInput &Input, const char *Cmd, HExecOutput &Output)
{
   char TmpStr[500 + 1];
   char FileName[500 + 1];
   bool MasterDownFlag = Input.MasterDownFlag;
   bool BackupLogFlag = false;
   bool RestoreFlag = Input.Func == efRestore;

   snprintf(ErrMsg, 8192, "Func = %d, FullName = %s, EnvName = %s, Cmd = %.300s", 
      Input.Func, Input.FullName, Input.EnvName, Cmd);
   Syslog.Log(__FILE__, __LINE__, llDebug, "ENTRY", ErrMsg);

   if ( ( ClusterFlag && LogLocation == llLocal ) ||
        ( MyRole == nrMaster && HotbackEnableFlag && !HotbackPausedFlag ) )
   {
      if ( Input.Func == efJob || Input.Func == efRestore )
      {
         BackupLogFlag = true;
      }
   }

   int PipeFd[2];

   if ( BackupLogFlag )
   {
      pipe(PipeFd);
   }

   snprintf(FileName, 500, "%s/log_%s/%s%s", LogDir.c_str(), Input.BatchTime,
      Input.FullName, RestoreFlag ? "_restore" : "");
   long OldSize = BkFileSize(FileName);
   pid_t Pid = fork();

   if ( Pid == -1 )
   {
      Output.Rc = -1;
      Output.Errno = errno;
      Output.Pid = -1;
      int fd = open(FileName, O_CREAT | O_WRONLY | O_APPEND | O_SYNC, 0666);
      if ( fd != -1 )
      {
         char TmpStr[500 + 1];
         snprintf(TmpStr, 500, "%s:%d fork fail, errno = %d\n", __FILE__, __LINE__, Output.Errno);
         write(fd, TmpStr, strlen(TmpStr));
         close(fd);
      }
      snprintf(ErrMsg, 8192, "Rc = %d, Errno = %d, Pid = %d", 
         Output.Rc, Output.Errno, Output.Pid);
      Syslog.Log(__FILE__, __LINE__, llDebug, "EXIT", ErrMsg);
      if ( BackupLogFlag )
      {
         BackupLog(Input.BatchTime, Input.FullName, OldSize, RestoreFlag);
         close(PipeFd[0]);
         close(PipeFd[1]);
      }
      sleep(10);
      return;
   }

   if ( Pid != 0 )
   {
      Output.Rc = 0;
      Output.Errno = 0;
      Output.Pid = Pid;
      if ( Input.Func == efJob && Input.DesServerId != MyId )
      {
         Output.Pid = 0;
      }
      snprintf(ErrMsg, 8192, "Rc = %d, Errno = %d, Pid = %d", 
         Output.Rc, Output.Errno, Output.Pid);
      Syslog.Log(__FILE__, __LINE__, llDebug, "EXIT", ErrMsg);
      if ( BackupLogFlag )
      {
         pid_t WorkerPid = fork();
         if ( WorkerPid == 0 )
         {
            close(PipeFd[1]);
            for ( ; ; )
            {
               char c;
               int r = BsReadFromSocket(PipeFd[0], &c, 1, 3000);
               if ( r == BS_ETIMEOUT )
               {
                  BackupLog(Input.BatchTime, Input.FullName, OldSize, RestoreFlag);
                  continue;
               }
               break;
            }
            BackupLog(Input.BatchTime, Input.FullName, OldSize, RestoreFlag);
            exit(0);
         }
         close(PipeFd[0]);
         close(PipeFd[1]);
      }
      return;
   }

   if ( BackupLogFlag )
   {
      close(PipeFd[0]);
   }

   close(SockExec);
   close(MsgSock);

   if ( Input.Func == efJob && Input.DitherMs > 0 )
   {
      snprintf(ErrMsg, 8192, "DitherMs = %d", Input.DitherMs);
      Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);
      BsSleep(Input.DitherMs);
   }

   if ( Input.Func == efJob )
   {
      pid_t ParentPid = getppid();
      if ( Input.SrcServerId != MyId )
      {
         HString Result;
         char AddrStr[200 + 1];
// 发 job_running
         int i;
         HClusterNodeInfo &Node = ClusterNodeVec[Input.SrcServerId];
         HString MasterAddr = Node.Addr;
         int MasterPort = Node.Port;
         snprintf(TmpStr, 500, "a|||%s|%ld|%d",
            Input.FullName, Input.JobStartTime, getpid());
         for ( i = 0; i < 60; i++ )
         {
            if ( MasterDownFlag && HotbackAddr != "" )
            {
               snprintf(AddrStr, 200, "%s:%d", HotbackAddr.c_str(), HotbackPort);
            }
            else
            {
               snprintf(AddrStr, 200, "%s:%d", MasterAddr.c_str(), MasterPort);
            }
            snprintf(ErrMsg, 8192, "CallServer(%s, JOB_RUNNING), Job = %s",
               AddrStr, Input.FullName);
            Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);
            time_t StartTime = time(NULL);
            bool r = CallServer(Syslog, COMM_TIME_OUT, AddrStr, "JOB_RUNNING", Input.FullName, TmpStr, true, Result, true);
            int DeltaTime = time(NULL) - StartTime;
            if ( DeltaTime > 3 )
            {
               snprintf(ErrMsg, 8192, "CallServer(%s, JOB_RUNNING), Job = %s, elapsed = %d",
                  AddrStr, Input.FullName, DeltaTime);
               Syslog.Log(__FILE__, __LINE__, llWarn, "ONERR", ErrMsg);
            }
            if ( !r )
            {
               sleep(10);
               MasterDownFlag = !MasterDownFlag;
               continue;
            }
            break;
         }
         if ( i == 60 )
         {
            snprintf(ErrMsg, 8192, "Send JOB_RUNNING(%s), retry %d times fail", Input.FullName, i);
            Syslog.Log(__FILE__, __LINE__, llError, "ON_ERROR", ErrMsg);
         }
      }
      int r, fd;
      bool FailFlag = false;
      ExitFlag = false;
      BkSignal(SIGCHLD, SIG_DFL, 0);
      BkSignal(SIGINT, SignalHandler, 0);
      setsid();
      close(0);
      open("/dev/null", O_RDONLY);
      close(1);
      snprintf(FileName, 500, "%s/log_%s/%s", LogDir.c_str(), Input.BatchTime, Input.FullName);
      fd = open(FileName, O_CREAT | O_WRONLY | O_APPEND, 0666);
      if ( fd == -1 )
      {
         fd = open("/dev/null", O_WRONLY);
      }
      dup2(1, 2);
      if ( Input.RerunFlag )
      {
         fprintf(stderr, "\n\n====== RERUN ======\n\n");
      }
      if ( Input.RestartFlag )
      {
         fprintf(stderr, "\n\n====== RESTART ======\n\n");
      }
      {
         BkPrintError("Command is :");
         const char *s = Cmd;
         for ( ; *s != '\0'; )
         {
            const char *s_end = strchr(s, ';');
            if ( s_end == NULL )
            {
               fprintf(stderr, "%s\n", s);
               break;
            }
            else
            {
               s_end++;
               fprintf(stderr, "%.*s\n", s_end - s, s);
               s = s_end;
            }
         }
         fprintf(stderr, "\n\n");
      }
      {
         char IdxStr[200 + 1];
         snprintf(IdxStr, 200, "%d", MyId + 1);
         setenv("SIMBAT_NODE_ID", IdxStr, 1);
         BkPrintError("SIMBAT_NODE_ID=%s", IdxStr);
         BkPrintError("SIMBAT_DATE=%s", GetEnv("SIMBAT_DATE").c_str());
         BkPrintError("SIMBAT_NEXT_DATE=%s", GetEnv("SIMBAT_NEXT_DATE").c_str());
         HClusterNodeInfo &Node = ClusterNodeVec[MyId];
         BkPrintError("IP_ADDR=%s:%d", Node.Addr.c_str(), Node.Port);
      }
      for ( int Idx = 0; ; )
      {
         if ( Input.LoopInterval > 0 )
         {
            char IdxStr[200 + 1];
            snprintf(IdxStr, 200, "%d", Idx + 1);
            setenv("SIMBAT_LOOP_INDEX", IdxStr, 1);
            BkPrintError("SIMBAT_LOOP_INDEX=%s", IdxStr);
         }
         {
            int Seq = StrToInt(Input.JobSeq + 8);
            char SeqStr[200 + 1];
            snprintf(SeqStr, 200, "%.8s", Input.JobSeq);
            setenv("SIMBAT_JOB_SEQ_DATE", SeqStr, 1);
            BkPrintError("SIMBAT_JOB_SEQ_DATE=%s", SeqStr);
            snprintf(SeqStr, 200, "%010d", Seq + Idx);
            setenv("SIMBAT_JOB_SEQ_SEQ", SeqStr, 1);
            BkPrintError("SIMBAT_JOB_SEQ_SEQ=%s", SeqStr);
            snprintf(SeqStr, 200, "%.8s%010d", Input.JobSeq, Seq + Idx);
            setenv("SIMBAT_JOB_SEQ", SeqStr, 1);
            BkPrintError("SIMBAT_JOB_SEQ=%s", SeqStr);
         }
         r = System(Cmd);
         if ( r == -1 || !WIFEXITED(r) || WEXITSTATUS(r) != 0 )
         {
            if ( r == -1 )
            {
               BkPrintError("System() fail, r = %d, errno = %d", r, errno);
            }
            else
            {
               if ( WIFEXITED(r) )
               {
                  if ( Input.LoopInterval > 0 && WEXITSTATUS(r) == 59 )
                  {
                     snprintf(TmpStr, 500, "c|||%s|%ld|%d|Y", Input.FullName, Input.JobStartTime, getpid());
                     BkPrintError("LOOP BREAK");
                     break;
                  }
                  BkPrintError("JOB EXITED, STATUS = %d", WEXITSTATUS(r));
               }
               else if ( WIFSTOPPED(r) )
               {
                  BkPrintError("JOB STOPPED, SIGNAL = %d", WSTOPSIG(r));
               }
               else if ( WIFSIGNALED(r) )
               {
                  BkPrintError("JOB TERMINATED, SIGNAL = %d", WTERMSIG(r));
               }
               else
               {
                  BkPrintError("JOB NOT STARTED, r = %d, errno = %d", r, errno);
               }
            }
            snprintf(TmpStr, 500, "d|||%s|%ld|%d", Input.FullName, Input.JobStartTime, getpid());
            FailFlag = true;
            break;
         }
         if ( Input.LoopInterval == 0 )
         {
            snprintf(TmpStr, 500, "c|||%s|%ld|%d|N", Input.FullName, Input.JobStartTime, getpid());
            BkPrintError("JOB COMPLETED SUCCESSFULLY");
            break;
         }
         time_t CurrTime = time(NULL);
         Idx++;
         if ( ( Input.LoopCount > 0 && Idx >= Input.LoopCount ) ||
              ( Input.LoopEndTime > 0 && CurrTime + Input.LoopInterval > Input.LoopEndTime ) )
         {
            snprintf(TmpStr, 500, "c|||%s|%ld|%d|N", Input.FullName, Input.JobStartTime, getpid());
            BkPrintError("LOOP COMPLETED");
            break;
         }
         if ( kill(ParentPid, 0) == -1 )
         {
            snprintf(TmpStr, 500, "d|||%s|%ld|%d", Input.FullName, Input.JobStartTime, getpid());
            FailFlag = true;
            BkPrintError("PARENT DOWN");
            break;
         }
         BkPrintError("sleep(%d)", Input.LoopInterval);
         sleep(Input.LoopInterval);
         if ( kill(ParentPid, 0) == -1 )
         {
            snprintf(TmpStr, 500, "d|||%s|%ld|%d", Input.FullName, Input.JobStartTime, getpid());
            FailFlag = true;
            BkPrintError("PARENT DOWN");
            break;
         }
         if ( ExitFlag )
         {
            snprintf(TmpStr, 500, "d|||%s|%ld|%d", Input.FullName, Input.JobStartTime, getpid());
            FailFlag = true;
            BkPrintError("USER CANCEL");
            break;
         }
      }
      {
         HString Result;
         char AddrStr[200 + 1];
         char AddrStr2[200 + 1];
         pid_t ParentPid = getppid();
// 发 job_success or job_fail
         int i = 0;
         bool MasterOk = false;
         bool SlaveOk = false;
         HClusterNodeInfo &Node = ClusterNodeVec[Input.SrcServerId];
         HString MasterAddr = Node.Addr;
         int MasterPort = Node.Port;
         {
            HClusterNodeInfo &Node = ClusterNodeVec[MyId];
            snprintf(AddrStr2, 200, "%s:%d", Node.Addr.c_str(), Node.Port);
         }
         if ( Input.SrcServerId == MyId )
         {
            SlaveOk = true;
         }
         for ( i = 0; i < 60; i++ )
         {
            if ( !MasterOk )
            {
               if ( MasterDownFlag && HotbackAddr != "" )
               {
                  snprintf(AddrStr, 200, "%s:%d", HotbackAddr.c_str(), HotbackPort);
               }
               else
               {
                  snprintf(AddrStr, 200, "%s:%d", MasterAddr.c_str(), MasterPort);
               }
               snprintf(ErrMsg, 8192, "CallServer(%s, %s), Job = %s",
                  AddrStr, FailFlag ? "JOB_FAIL" : "JOB_SUCCESS", Input.FullName);
               Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);
               bool r = CallServer(Syslog, COMM_TIME_OUT, AddrStr,
                  FailFlag ? "JOB_FAIL" : "JOB_SUCCESS", Input.FullName, TmpStr, true, Result, true);
               if ( r )
               {
                  MasterOk = true;
               }
               else
               {
                  MasterDownFlag = !MasterDownFlag;
               }
            }
            if ( !SlaveOk )
            {
               snprintf(ErrMsg, 8192, "CallServer(%s, %s), Job = %s",
                  AddrStr2, FailFlag ? "JOB_FAIL" : "JOB_SUCCESS", Input.FullName);
               Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);
               bool r = CallServer(Syslog, COMM_TIME_OUT, AddrStr2,
                  FailFlag ? "JOB_FAIL" : "JOB_SUCCESS", Input.FullName, TmpStr, true, Result, true);
               if ( r )
               {
                  SlaveOk = true;
               }
            }
            if ( MasterOk && SlaveOk )
            {
               break;
            }
            sleep(10);
         }
         if ( i == 60 )
         {
            snprintf(ErrMsg, 8192, "Send %s(%s), retry %d times fail",
               FailFlag ? "JOB_FAIL" : "JOB_SUCCESS", Input.FullName, i);
            Syslog.Log(__FILE__, __LINE__, llError, "ON_ERROR", ErrMsg);
         }
      }
      fflush(stdout);
      fflush(stderr);
      fsync(fd);
      exit(0);
   }

   if ( Input.Func == efRestore )
   {
      int r, fd;
      bool FailFlag = false;
      BkSignal(SIGCHLD, SIG_DFL, 0);
      BkSignal(SIGINT, SIG_IGN, 0);
      setsid();
      close(0);
      open("/dev/null", O_RDONLY);
      close(1);
      snprintf(FileName, 500, "%s/log_%s/%s_restore", LogDir.c_str(), Input.BatchTime, Input.FullName);
      fd = open(FileName, O_CREAT | O_WRONLY | O_APPEND, 0666);
      if ( fd == -1 )
      {
         fd = open("/dev/null", O_WRONLY);
      }
      dup2(1, 2);
      fprintf(stderr, "\n\n====== RESTORE ======\n\n");
      BkPrintError("Command is :");
      const char *s = Cmd;
      for ( ; *s != '\0'; )
      {
         const char *s_end = strchr(s, ';');
         if ( s_end == NULL )
         {
            fprintf(stderr, "%s\n", s);
            break;
         }
         else
         {
            s_end++;
            fprintf(stderr, "%.*s\n", s_end - s, s);
            s = s_end;
         }
      }
      {
         int Seq = StrToInt(Input.JobSeq + 8);
         char SeqStr[200 + 1];
         snprintf(SeqStr, 200, "%.8s", Input.JobSeq);
         setenv("SIMBAT_JOB_SEQ_DATE", SeqStr, 1);
         BkPrintError("SIMBAT_JOB_SEQ_DATE=%s", SeqStr);
         snprintf(SeqStr, 200, "%010d", Seq);
         setenv("SIMBAT_JOB_SEQ_SEQ", SeqStr, 1);
         BkPrintError("SIMBAT_JOB_SEQ_SEQ=%s", SeqStr);
         snprintf(SeqStr, 200, "%.8s%010d", Input.JobSeq, Seq);
         setenv("SIMBAT_JOB_SEQ", SeqStr, 1);
         BkPrintError("SIMBAT_JOB_SEQ=%s", SeqStr);
      }
      r = System(Cmd);
      if ( r == -1 || !WIFEXITED(r) || WEXITSTATUS(r) != 0 )
      {
         if ( r == -1 )
         {
            BkPrintError("System() fail, r = %d, errno = %d", r, errno);
         }
         else
         {
            if ( WIFEXITED(r) )
            {
               BkPrintError("JOB EXITED, STATUS = %d", WEXITSTATUS(r));
            }
            else if ( WIFSTOPPED(r) )
            {
               BkPrintError("JOB STOPPED, SIGNAL = %d", WSTOPSIG(r));
            }
            else if ( WIFSIGNALED(r) )
            {
               BkPrintError("JOB TERMINATED, SIGNAL = %d", WTERMSIG(r));
            }
            else
            {
               BkPrintError("JOB NOT STARTED, r = %d, errno = %d", r, errno);
            }
         }
         snprintf(TmpStr, 500, "f|||%s", Input.FullName);
         FailFlag = true;
      }
      else
      {
         snprintf(TmpStr, 500, "e|||%s", Input.FullName);
      }
      {
         HString Result;
         char AddrStr[200 + 1];
         pid_t ParentPid = getppid();
// 发 restore_success or restore_fail
         int i;
         HClusterNodeInfo &Node = ClusterNodeVec[Input.SrcServerId];
         snprintf(AddrStr, 200, "%s:%d", "127.0.0.1", Node.Port);
         for ( i = 0; i < 60; i++ )
         {
            snprintf(ErrMsg, 8192, "CallServer(%s, %s), Job = %s",
               AddrStr, FailFlag ? "RESTORE_FAIL" : "RESTORE_SUCCESS", Input.FullName);
            Syslog.Log(__FILE__, __LINE__, llDebug, "OTHER", ErrMsg);
            bool r = CallServer(Syslog, COMM_TIME_OUT, AddrStr,
               FailFlag ? "RESTORE_FAIL" : "RESTORE_SUCCESS", Input.FullName, TmpStr, true, Result, true);
            if ( !r )
            {
               sleep(10);
               continue;
            }
            break;
         }
         if ( i == 60 )
         {
            snprintf(ErrMsg, 8192, "Send %s(%s), retry %d times fail",
               FailFlag ? "RESTORE_FAIL" : "RESTORE_SUCCESS", Input.FullName, i);
            Syslog.Log(__FILE__, __LINE__, llError, "ON_ERROR", ErrMsg);
         }
      }
      fflush(stdout);
      fflush(stderr);
      fsync(fd);
      exit(0);
   }

   if ( Input.Func == efTrigger )
   {
      System2(Cmd);
   }

   exit(0);
}

void DoIt()
{
   pid_t ParentPid = getppid();
   bool ExitFlag = false;

   for ( ; !ExitFlag; )
   {
      int r;
      MsgSock = BsAcceptConnectionUD(SockExec, NULL, 10);
      if ( MsgSock < 0 && MsgSock != BS_ETIMEOUT )
      {
         snprintf(ErrMsg, 8192, "BsAcceptConnection fail, r = %d, errno = %d", MsgSock, errno);
         Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
         exit(8);
      }
      if ( kill(ParentPid, 0) == -1 )
      {
         snprintf(ErrMsg, 8192, "parent down");
         Syslog.Log(__FILE__, __LINE__, llWarn, "OTHER", ErrMsg);
         ExitFlag = true;
      }
      if ( MsgSock == BS_ETIMEOUT )
      {
         continue;
      }
      HExecInput Input;
      HExecOutput Output;
      memset(&Output, 0, sizeof(Output));
      r = BsReadFromSocket(MsgSock, &Input, sizeof(Input), COMM_TIME_OUT);
      if ( r != sizeof(Input) )
      {
         snprintf(ErrMsg, 8192, "ReadFromSocket fail, r = %d, errno = %d", r, errno);
         Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
         exit(8);
      }
      char Command[8192 + 1];
      if ( Input.CmdStrLen > sizeof(Command) - 1 )
      {
         Output.Rc = -1;
         Output.Errno = ENOMEM;
      }
      else
      {
         r = BsReadFromSocket(MsgSock, Command, Input.CmdStrLen, COMM_TIME_OUT);
         if ( r != Input.CmdStrLen )
         {
            snprintf(ErrMsg, 8192, "ReadFromSocket fail, r = %d, errno = %d", r, errno);
            Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
            exit(8);
         }
         Command[Input.CmdStrLen] = '\0';
         if ( Input.Func == efReloadConfig )
         {
            HString ErrStr;
            if ( !LoadCfg(ErrStr) )
            {
               snprintf(ErrMsg, 8192, "Load config.xml fail, %s", ErrStr.c_str());
               Syslog.Log(__FILE__, __LINE__, llError, "ONERR", ErrMsg);
            }
         }
         else if ( Input.Func == efRefreshVarpool )
         {
            LoadVarpool();
         }
         else if ( Input.Func == efSetEnv )
         {
            SetEnv(Input.EnvName, Input.FullName);
         }
         else
         {
            RunIt(Input, Command, Output);
         }
      }
      r = BsWriteToSocket(MsgSock, &Output, sizeof(Output), COMM_TIME_OUT);
      if ( r != 0 )
      {
         snprintf(ErrMsg, 8192, "BsWriteToSocket fail, r = %d, errno = %d", r, errno);
         Syslog.Log(__FILE__, __LINE__, errno == EPIPE ? llWarn : llError, "ONERR", ErrMsg);
      }
      close(MsgSock);
   }
}

int main(int argc, char *argv[])
{
   if ( argc != 2 )
   {
      BkPrintError("Line %d : argc != 2", __LINE__);
      return 8;
   }

   SockExec = StrToInt(argv[1]);

   SimbatBase = getenv("SIMBAT_BASE");

   if ( SimbatBase == NULL )
   {
      BkPrintError("Line %d : $SIMBAT_BASE not defined", __LINE__);
      return 8;
   }

   HString ErrStr;

   if ( !LoadCfg(ErrStr) )
   {
      BkPrintError("Load config.xml fail, %s", ErrStr.c_str());
      return 8;
   }

   char FileName[500 + 1];
   snprintf(FileName, 500, "%s/status/var_pool", ShrDir.c_str());
   VarpoolFile = FileName;
   snprintf(FileName, 500, "%s/tmp/var_pool.lck", SimbatBase);
   VarpoolLockFile = FileName;
   LoadVarpool();
   BkSignalIgnore(SIGPIPE);
   BkSignalIgnore(SIGCHLD);
   DoIt();
   return 0;
}
